package com.shiguiwu.springmybatis.javabase.thread;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @description: 线程池
 * @author: stone
 * @date: Created by 2021/6/24 15:21
 * @version: 1.0.0
 * @pakeage: com.shiguiwu.springmybatis.thread
 */
@Slf4j
public class ExecutorTests {

    public static AtomicInteger threadNum = new AtomicInteger(1);


    public static void main(String[] args) throws InterruptedException {
//        testSynchronousQueue();
//        testPriorityBlockingQueue();
        testThreadFactory();
//        testRejectedExecutionHandler();
//        extendsExecutor();
    }

    /**
     * 从输出中可以看出，系统创建了50个线程处理任务，代码中使用了SynchronousQueue同步队列，这种队列比较特殊，
     * 放入元素必须要有另外一个线程去获取这个元素，否则放入元素会失败或者一直阻塞在那里直到有线程取走，示例中任务处理休眠了指定的时间，
     * 导致已创建的工作线程都忙于处理任务，所以新来任务之后，将任务丢入同步队列会失败，丢入队列失败之后，会尝试新建线程处理任务。
     * 使用上面的方式创建线程池需要注意，如果需要处理的任务比较耗时，会导致新来的任务都会创建新的线程进行处理，可能会导致创建非常多的线程，
     * 最终耗尽系统资源，触发OOM。
     */
    public static void testSynchronousQueue() {
        ExecutorService service = Executors.newCachedThreadPool();
        for (int i = 0; i < 50; i++) {
            int j = i;
            service.execute(() -> {
                System.out.println(Thread.currentThread().getName() + "处理任务" + j);
                try {
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            });
        }
        service.shutdown();
    }

    //优先级
    public static void testPriorityBlockingQueue() {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                1,
                1,
                60L,
                TimeUnit.SECONDS,
                new PriorityBlockingQueue()
        );
        for (int i = 0; i < 10; i++) {
            executor.execute(new Task("task" + i, +i));
        }

        for (int i = 100; i >= 90; i--) {
            executor.execute(new Task("task" + i, +i));
        }

        executor.shutdown();

    }

    //自定义线程池
    public static void testThreadFactory() {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                5,
                5,
                60L,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(100),
                //线程工厂
                (r) -> {
                    Thread t = new Thread(r);
                    t.setName("自定义线程-" + threadNum.getAndDecrement());
                    return t;
                }
        );



        executor.setCorePoolSize(6);
        executor.setMaximumPoolSize(100);
        for (int i = 0; i < 1000; i++) {
            String taskName = "任务" + i;
            executor.execute(() -> {
                log.info(Thread.currentThread().getName() + "处理" + taskName);

            });
        }
        //可以设置，初始化后什么都可以设置了，如果最大线程数设置比核心小，则抛异常
        //executor.setCorePoolSize(1);
        //executor.setMaximumPoolSize(6);//报错
        log.info("====================================>");
        executor.shutdown();
    }

    public static void testRejectedExecutionHandler() {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                1,
                1,
                60L,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1),
                new MyRejectedExecutionHandler()
        );

        for (int i = 0; i < 5; i++) {
            executor.execute(() -> {
                log.info(Thread.currentThread().getName() + "处理");
                try {
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            });
        }
        executor.shutdown();
    }

    //线程池扩展，执行之前，执行之后
    public static void extendsExecutor() throws InterruptedException {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                10,
                10,
                60L,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(1),
                Executors.defaultThreadFactory(),
                new MyRejectedExecutionHandler()
        ) {
            @Override
            protected void beforeExecute(Thread t, Runnable r) {
                System.out.println(t.getName() + "=======================>start");
            }

            @Override
            protected void afterExecute(Runnable r, Throwable t) {
                System.out.println(r.getClass().getName() + "=======================>end");
            }

            @Override
            protected void terminated() {
                System.out.println("end ====================================>");
            }
        };


        for (int i = 0; i < 10; i++) {
            executor.execute(() -> {
                System.out.println(Thread.currentThread().getName() + "处理");
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            });
        }
        TimeUnit.SECONDS.sleep(1);
        executor.shutdown();
    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Task implements Runnable, Comparable<Task> {

        private String name;

        private int age;

        @Override
        public int compareTo(Task o) {
            return Integer.compare(o.age, this.age);
        }

        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName() + "处理" + this.name);
        }
    }

    //自定义饱和策略
    public static class MyRejectedExecutionHandler implements RejectedExecutionHandler {

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            //自定义饱和策略
            //记录一下无法处理的任务
            System.out.println("无法处理的任务：" + r.toString());
        }
    }


}
